package com.iov.failurelog.utils;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.MessageFormat;
import java.util.*;


@Service
public class HbaseTemplate {

    private Connection connection;

    /**
     * 值过滤器
     * ValueFilter：值过滤器，找到符合值条件的键值对
     * 模糊查找：scan 'test', FILTER => "ValueFilter(=,'substring:张')"
     * 说明：查询test表中，所有列值为以“张”开头的数据
     * 精确查询：scan 'test', FILTER => "ValueFilter(=,'binary:张三')"
     * 说明：查询test表中，所有列值为“张三”的数据
     * 注意1：ValueFilter 过滤器可以利用 get 和 scan 方法对单元格进行过滤，但是使用 get 方法时，需要指定行键
     *
     * @param tableName
     * @param keyword
     * @return
     */
    public Map<String, Map<String, String>> getResultScannerValueFilter(
            String tableName, String keyword) {
        Scan scan = new Scan();
        if (StringUtils.isNotBlank(keyword)) {
            Filter filter = new ValueFilter(CompareOperator.GREATER_OR_EQUAL, new SubstringComparator(keyword));
            scan.setFilter(filter);
        }
        return this.queryData(tableName, scan);
    }

    /**
     * 根据时间区间进行查询
     *
     * @param tableName
     * @param beginTime
     * @param endTime
     * @return
     * @throws IOException
     */
    public Map<String, Map<String, String>> getResultScannerByTimeRange(
            String tableName,
            Date beginTime, Date endTime) throws IOException {
        Scan scan = new Scan();
        scan.setTimeRange(beginTime.getTime(), endTime.getTime());
        return this.queryData(tableName, scan);
    }


    /**
     * 通过表名以及过滤条件查询数据
     *
     * @param tableName 表名
     * @param scan      过滤条件
     * @return java.util.Map<java.lang.String, java.util.Map < java.lang.String, java.lang.String>>
     * @author
     * @date 2018/7/4 16:13
     * @since 1.0.0
     */
    private Map<String, Map<String, String>> queryData(String tableName, Scan scan) {
        //<rowKey,对应的行数据>
        Map<String, Map<String, String>> result = new HashMap<>();
        ResultScanner rs = null;
        // 获取表
        Table table = null;
        try {
            table = getTable(tableName);
            rs = table.getScanner(scan);
            for (Result r : rs) {
                //每一行数据
                Map<String, String> columnMap = new HashMap<>();
                String rowKey = null;
                for (Cell cell : r.listCells()) {
                    if (rowKey == null) {
                        rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                    }
                    columnMap.put(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()), Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                }

                if (rowKey != null) {
                    result.put(rowKey, columnMap);
                }
            }
        } catch (IOException e) {
            System.err.println(MessageFormat.format("遍历查询指定表中的所有数据失败,tableName:{0}"
                    , tableName));
            e.printStackTrace();
        } finally {
            close(null, rs, table);
        }

        return result;
    }

    /**
     * 为表的某个单元格赋值
     *
     * @param tableName  表名
     * @param rowKey     rowKey
     * @param familyName 列族名
     * @param column1    列名
     * @param value1     列值
     * @author
     * @date 2018/7/4 10:20
     * @since 1.0.0
     */
    public void setColumnValue(String tableName,
                               String rowKey,
                               String familyName,
                               String column1, String value1) {
        Table table = null;
        try {
            // 获取表
            table = getTable(tableName);
            // 设置rowKey
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column1), Bytes.toBytes(value1));

            table.put(put);
            System.out.println("add data Success!");
        } catch (IOException e) {
            System.err.println(MessageFormat.format("为表的某个单元格赋值失败,tableName:{0},rowKey:{1},familyName:{2},column:{3}"
                    , tableName, rowKey, familyName, column1));
            e.printStackTrace();
        } finally {
            close(null, null, table);
        }
    }

    /**
     * 删除指定的单元格
     *
     * @param tableName  表名
     * @param rowKey     rowKey
     * @param familyName 列族名
     * @param columnName 列名
     * @return boolean
     * @author
     * @date 2018/7/4 11:41
     * @since 1.0.0
     */
    public boolean deleteColumn(String tableName, String rowKey,
                                String familyName, String columnName) {
        Table table = null;
        Admin admin = null;
        try {
            admin = connection.getAdmin();

            if (admin.tableExists(TableName.valueOf(tableName))) {
                // 获取表
                table = getTable(tableName);
                Delete delete = new Delete(Bytes.toBytes(rowKey));
                // 设置待删除的列
                delete.addColumns(Bytes.toBytes(familyName), Bytes.toBytes(columnName));

                table.delete(delete);
                System.out.println(MessageFormat.format("familyName({0}):columnName({1})is deleted!", familyName, columnName));
            }

        } catch (IOException e) {
            System.err.println(MessageFormat.format("删除指定的列失败,tableName:{0},rowKey:{1},familyName:{2},column:{3}"
                    , tableName, rowKey, familyName, columnName));
            e.printStackTrace();
            return false;
        } finally {
            close(admin, null, table);
        }
        return true;
    }

    /**
     * 根据rowKey删除指定的行
     *
     * @param tableName 表名
     * @param rowKey    rowKey
     * @return boolean
     * @author
     * @date 2018/7/4 13:26
     * @since 1.0.0
     */
    public boolean deleteRow(String tableName,
                             String rowKey) {
        Table table = null;
        Admin admin = null;
        try {
            admin = connection.getAdmin();

            if (admin.tableExists(TableName.valueOf(tableName))) {
                // 获取表
                table = getTable(tableName);
                Delete delete = new Delete(Bytes.toBytes(rowKey));

                table.delete(delete);
                System.out.println(MessageFormat.format("row({0}) is deleted!", rowKey));
            }
        } catch (IOException e) {
            System.err.println(MessageFormat.format("删除指定的行失败,tableName:{0},rowKey:{1}"
                    , tableName, rowKey));
            e.printStackTrace();
            return false;
        } finally {
            close(admin, null, table);
        }
        return true;
    }

    /**
     * 根据columnFamily删除指定的列族
     *
     * @param tableName    表名
     * @param columnFamily 列族
     * @return boolean
     * @author
     * @date 2018/7/4 13:26
     * @since 1.0.0
     */
    public boolean deleteColumnFamily(String tableName, String columnFamily) {
        Admin admin = null;
        try {
            admin = connection.getAdmin();

            if (admin.tableExists(TableName.valueOf(tableName))) {
                admin.deleteColumnFamily(TableName.valueOf(tableName), Bytes.toBytes(columnFamily));
                System.out.println(MessageFormat.format("familyName({0}) is deleted!", columnFamily));
            }
        } catch (IOException e) {
            System.err.println(MessageFormat.format("删除指定的列族失败,tableName:{0},columnFamily:{1}"
                    , tableName, columnFamily));
            e.printStackTrace();
            return false;
        } finally {
            close(admin, null, null);
        }
        return true;
    }

    /**
     * 获取table
     *
     * @param tableName 表名
     * @return Table
     * @throws IOException IOException
     */
    private Table getTable(String tableName) throws IOException {
        return connection.getTable(TableName.valueOf(tableName));
    }

    /**
     * 关闭流
     */
    private void close(Admin admin, ResultScanner rs, Table table) {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                System.err.println("关闭Admin失败");
                e.printStackTrace();
            }
        }

        if (rs != null) {
            rs.close();
        }

        if (table != null) {
            try {
                table.close();
            } catch (IOException e) {
                System.err.println("关闭Table失败 " + e.getMessage());
            }
        }
    }

    /**
     * 遍历查询指定表中的  所有数据
     * @author
     * @date 2018/7/3 18:21
     * @since 1.0.0
     * @param tableName 表名
     * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
     */
    public Map<String, Map<String, String>> getResultScanner(String tableName){
        Scan scan = new Scan();
        return this.queryData(tableName,scan);
    }

}